package com.cjy.homework;

import com.google.common.base.Throwables;
import com.google.gson.Gson;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.GregorianCalendar;

public class SumData {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf =  new Configuration();
        conf.set("fs.defaultFS","hdfs://20210322059-master:9000");

        SimpleDateFormat tempDate = new SimpleDateFormat("yyyy-MM-dd");
        Calendar calendar = new GregorianCalendar();
        calendar.add(calendar.DATE, -1);
        String tomorrow = tempDate.format(calendar.getTime());
        String input = "/mydata/"+tomorrow+"/"+tomorrow+".json";
        String out = "/mydata2/total_num/"+tomorrow;

        Job job = Job.getInstance(conf,"json to mysql");
        job.setJarByClass(SumData.class);
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.addInputPath(job,new Path(input));
        FileOutputFormat.setOutputPath(job,new Path(out));

        job.waitForCompletion(true);
    }
    static class MyMapper extends Mapper<LongWritable, Text,Text, IntWritable>{
        @Override
        protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException {
            Gson gson = new Gson();
            Datas2 datas = gson.fromJson(String.valueOf(value),Datas2.class);
            String s = datas.collectTime;
            context.write(new Text(s),new IntWritable(1));

        }
    }
    static class MyReducer extends Reducer<Text,IntWritable, NullWritable,NullWritable>{
        private Connection conn;
        public MyReducer() {
            try {
                Class.forName("com.mysql.cj.jdbc.Driver");
                String url = "jdbc:mysql://192.168.65.3:3306/mydata?serverTimezone=UTC&characterEncoding=utf-8&useSSL=false";
                this.conn = DriverManager.getConnection(url,"root","Aa123-456");
            }catch (Throwable e){
                System.out.println("######MyReducer Throwable....");
                System.out.println(Throwables.getStackTraceAsString(e));
            }
        }

        @Override
        protected void cleanup(Reducer<Text, IntWritable, NullWritable, NullWritable>.Context context) throws IOException, InterruptedException {
            try {
                System.out.println("######MyReducer cleanup....");
                this.conn.close();
            }catch (Throwable e){
                System.out.println(Throwables.getStackTraceAsString(e));
            }
        }

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            System.out.println("######MyReducer reduce....");
            int sum=0;
            for (IntWritable val:values){
                sum += val.get();
            }
            String sql = "insert into `total_num` values(?,?)";
            try {
                PreparedStatement preparedStatement = this.conn.prepareStatement(sql);
                preparedStatement.setString(1, String.valueOf(key));
                preparedStatement.setInt(2,sum);
                System.out.println("######MyReducer insert....");
                preparedStatement.execute();
            }catch (Throwable e){
                System.out.println("######MyReducer reduce....");
                System.out.println(Throwables.getStackTraceAsString(e));
            }
        }
    }
    public static class Datas2 {
        String collectTime;
        String vin;
        public Datas2(String collectTime, String vin) {
            this.collectTime = collectTime;
            this.vin = vin;
        }
    }
}
